package com.longge.common.util;

import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.ProtectACLCreateModeStatPathAndBytesable;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import com.longge.common.dto.ZkNodeDto;

import lombok.NonNull;

public class ZkUtils {
	private ZkUtils() {}
	
	private static CuratorFramework client;
	private static Map<String, PathChildrenCache> childrenCaches = new ConcurrentHashMap<>();
	private static Map<String, NodeCache> nodeCaches = new ConcurrentHashMap<>();

	public static void setClient(CuratorFramework client) {
		ZkUtils.client = client;
	}
	
	/**
	 * add zk node
	 * @param path
	 * @param data
	 * @param mode
	 * @return
	 */
	public static Stat addNode(@NonNull String path, byte[] data, @NonNull CreateMode mode) {
		try {
			Stat stat = new Stat();
			ProtectACLCreateModeStatPathAndBytesable<String> build = client.create().creatingParentsIfNeeded();
			if(null != mode) {
				build.withMode(mode);
			}
			build.storingStatIn(stat);
			if(data != null && data.length > 0) {
				build.forPath(path, data);
			} else {
				build.forPath(path);
			}
			return stat;
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}
	
	/**
	 * delete node, if node has children node / node not exits,
	 * @param path
	 */
	public static void deleteNode(@NonNull String path) {
		// if node has children node, delete is error
		try {
			client.delete().forPath(path);
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}
	
	/**
	 * delete node and children node, if node not exits,
	 * @param path
	 */
	public static void deleteNodeAndChildrenNode(@NonNull String path) {
		// if node has children node, delete all children and this node
		try {
			client.delete().deletingChildrenIfNeeded().forPath(path);
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}
	
	/**
	 * update node
	 * @param path
	 * @param data
	 * @return
	 */
	public static Stat updateNode(@NonNull String path, byte[] data) {
		try {
			if(data != null && data.length > 0) {
				return client.setData().forPath(path, data);
			}
			return client.setData().forPath(path);
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}
	
	/**
	 * update node with version
	 * @param path
	 * @param data
	 * @param version
	 * @return
	 */
	public static Stat updateNodeWithVersion(@NonNull String path, byte[] data, @NonNull Integer version) {
		try {
			if(data != null && data.length > 0) {
				return client.setData().withVersion(version).forPath(path, data);
			}
			return client.setData().withVersion(version).forPath(path);
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}
	
	/**
	 * check node exists
	 * @param path
	 * @return
	 * @throws Exception
	 */
	public static Boolean checkExists(@NonNull String path) {
		try {
			// check exists node
			Stat stat = client.checkExists().forPath(path);
			if (null == stat) {
				return Boolean.FALSE;
			}
			return Boolean.TRUE;
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}
	
	/**
	 * list children nodes
	 * @param path
	 * @return
	 * @throws Exception
	 */
	public static List<ZkNodeDto> listChildrenNodes(@NonNull String path) {
		try {
			Stat pStat = new Stat();
			List<String> nodes = client.getChildren().storingStatIn(pStat).forPath(path);
			if(null!=pStat && pStat.getNumChildren()>0) {
				return nodes.parallelStream().map(item -> {
					ZkNodeDto dto = ZkNodeDto.builder()
							.path(getChildrenNodePath(path, item))
							.build();
					dto.setRealPath(getRealPath(dto.getPath()));
					
					// only get node data
					// byte[] data = client.getData().forPath(dto.getPath());
					Stat stat = new Stat();
					byte[] data = null;
					try {
						data = client.getData().storingStatIn(stat).forPath(dto.getPath());
					} catch (Exception e) {
						throw new RuntimeException(e);
					}
					if (null != data && data.length > 0) {
						dto.setData(new String(data));
					}
					dto.setStat(stat);
					dto.setIsTempNode(stat.getEphemeralOwner() > 0);
					return dto;
				}).collect(Collectors.toList());
				
			}
			return Collections.emptyList();
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}
	
	/**
	 * init PathChildrenCache if cache not exists
	 * @param parentPath
	 * @return PathChildrenCache
	 */
	public static PathChildrenCache initAndGetPathChildrenCacheIfNeed(@NonNull String parentPath) {
		if(!childrenCaches.containsKey(parentPath)) {
			synchronized (parentPath) {
				if(!childrenCaches.containsKey(parentPath)) {
					PathChildrenCache childrenCache = new PathChildrenCache(client, parentPath, true);
					childrenCaches.put(parentPath, childrenCache);
					/**
					 * StartMode: 初始化方式 
					 * POST_INITIALIZED_EVENT：异步初始化，初始化之后会触发事件
					 * NORMAL：异步初始化
					 * BUILD_INITIAL_CACHE：同步初始化
					 */
					try {
						childrenCache.start(StartMode.BUILD_INITIAL_CACHE);
					} catch (Exception e) {
						throw new RuntimeException(e);
					}
				}
			}
		}
		return childrenCaches.get(parentPath);
	}
	
	/**
	 * add children node listener
	 * @param parentPath
	 * @param listener
	 */
	public static void addChildrenListener(@NonNull String parentPath, @NonNull PathChildrenCacheListener listener) {
		initAndGetPathChildrenCacheIfNeed(parentPath).getListenable().addListener(listener);
	}
	
	/**
	 * remove children node listener
	 * @param parentPath
	 * @param listener
	 */
	public static void removeChildrenListener(@NonNull String parentPath, @NonNull PathChildrenCacheListener listener) {
		childrenCaches.get(parentPath).getListenable().removeListener(listener);
	}
	
	/**
	 * remove all children node listener
	 * @param parentPath
	 */
	public static void clearChildrenListener(@NonNull String parentPath) {
		childrenCaches.get(parentPath).getListenable().clear();;
	}
	
	/**
	 * get NodeCache, if not exists, init it
	 * @param nodePath
	 * @return
	 */
	public static NodeCache initAndGetNodeCacheIfNeed(@NonNull String nodePath) {
		if(!nodeCaches.containsKey(nodePath)) {
			synchronized (nodePath) {
				if(!nodeCaches.containsKey(nodePath)) {
					NodeCache nowCache = new NodeCache(client, nodePath);
					nodeCaches.put(nodePath, nowCache);
					try {
						nowCache.start();
					} catch (Exception e) {
						throw new RuntimeException(e);
					}
				}
			}
		}
		return nodeCaches.get(nodePath);
	}
	
	/**
	 * add one node listener
	 * @param nodePath
	 * @param listener
	 */
	public static void addNodeListener(@NonNull String nodePath, @NonNull NodeCacheListener listener) {
		initAndGetNodeCacheIfNeed(nodePath).getListenable().addListener(listener);
	}
	
	/**
	 * remove one node listener
	 * @param nodePath
	 * @param listener
	 */
	public static void removeNodeListener(@NonNull String nodePath, @NonNull NodeCacheListener listener) {
		nodeCaches.get(nodePath).getListenable().removeListener(listener);
	}
	
	/**
	 * remove one node's all listener
	 * @param nodePath
	 */
	public static void clearNodeListener(@NonNull String nodePath) {
		nodeCaches.get(nodePath).getListenable().clear();
	}
	
	private static String getChildrenNodePath(String basePath, String childrenNode) {
		if (basePath.endsWith("/")) {
			return basePath + childrenNode;
		}
		return basePath + "/" + childrenNode;
	}

	private static String getRealPath(String path) {
		if (StringUtils.isBlank(client.getNamespace())) {
			return path;
		}
		return "/" + client.getNamespace() + path;
	}
}
